package org.tao.lightningmq.broker.netty.broker;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.*;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.BrokerEventCode;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.event.model.BaseEvent;
import org.tao.lightningmq.common.remote.SyncFuture;
import org.tao.lightningmq.common.remote.SyncFutureManager;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.net.InetSocketAddress;
import java.util.UUID;

/**
 * @Author lht
 * @date 2025/1/19 - 22:54
 * @description:
 */
@ChannelHandler.Sharable
public class BrokerServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger LOGGER = LoggerFactory.getLogger(BrokerServerHandler.class);

    private final EventBus eventBus;

    public BrokerServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        this.eventBus.init();
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        BaseEvent event = null;
        if (BrokerEventCode.REGISTRY_CONSUMER_GROUP.getCode() == code) {
            RegistryConsumerGroupReqDTO reqDTO = JSON.parseObject(body, RegistryConsumerGroupReqDTO.class);
            InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
            reqDTO.setIp(inetSocketAddress.getHostString());
            reqDTO.setPort(inetSocketAddress.getPort());
            event = new RegistryConsumerGroupEvent(reqDTO);
            event.setMsgId(reqDTO.getMsgId());
        } else if (BrokerEventCode.PUSH_MSG.getCode() == code) {
            MessageDTO messageDTO = JSON.parseObject(body, MessageDTO.class);
            checkTopicIsCreate(messageDTO.getTopic(), channelHandlerContext);
            PushMsgEvent pushMsgEvent = new PushMsgEvent();
            pushMsgEvent.setMessageDTO(messageDTO);
            pushMsgEvent.setMsgId(messageDTO.getMsgId());
            LOGGER.info("收到消息推送内容:{},message is {}", new String(messageDTO.getBody()), JSON.toJSONString(messageDTO));
            event = pushMsgEvent;
        } else if (BrokerEventCode.CONSUME_MSG.getCode() == code) {
            ConsumeMsgReqDTO consumeMsgReqDTO = JSON.parseObject(body, ConsumeMsgReqDTO.class);
            InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
            consumeMsgReqDTO.setIp(inetSocketAddress.getHostString());
            consumeMsgReqDTO.setPort(inetSocketAddress.getPort());
            ConsumeMsgEvent consumeMsgEvent = new ConsumeMsgEvent();
            consumeMsgEvent.setConsumeMsgReqDTO(consumeMsgReqDTO);
            consumeMsgEvent.setMsgId(consumeMsgReqDTO.getMsgId());
            channelHandlerContext.attr(AttributeKey.valueOf("consumer-reqId"))
                    .set(consumeMsgReqDTO.getReqId());
            event = consumeMsgEvent;
        } else if (BrokerEventCode.CONSUME_SUCCESS_MSG.getCode() == code) {
            ConsumeMsgAckReqDTO consumeMsgAckReqDTO = JSON.parseObject(body, ConsumeMsgAckReqDTO.class);
            InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
            consumeMsgAckReqDTO.setIp(inetSocketAddress.getHostString());
            consumeMsgAckReqDTO.setPort(inetSocketAddress.getPort());
            ConsumeMsgAckEvent consumeMsgAckEvent = new ConsumeMsgAckEvent();
            consumeMsgAckEvent.setConsumeMsgAckReqDTO(consumeMsgAckReqDTO);
            consumeMsgAckEvent.setMsgId(consumeMsgAckReqDTO.getMsgId());
            event = consumeMsgAckEvent;
        } else if (BrokerEventCode.CREATE_TOPIC.getCode() == code) {
            CreateTopicReqDTO createTopicReqDTO = JSON.parseObject(body, CreateTopicReqDTO.class);
            CreateTopicEvent createTopicEvent = new CreateTopicEvent();
            createTopicEvent.setCreateTopicReqDTO(createTopicReqDTO);
            createTopicEvent.setMsgId(createTopicReqDTO.getMsgId());
            event = createTopicEvent;
        } else if (BrokerEventCode.START_SYNC_MSG.getCode() == code) {
            StartSyncReqDTO startSyncReqDTO = JSON.parseObject(body, StartSyncReqDTO.class);
            StartSyncEvent startSyncEvent = new StartSyncEvent();
            startSyncEvent.setMsgId(startSyncReqDTO.getMsgId());
            event = startSyncEvent;
        } else if (BrokerResponseCode.SLAVE_SYNC_RESP.getCode() == code) {
            SlaveSyncRespDTO slaveSyncRespDTO = JSON.parseObject(body, SlaveSyncRespDTO.class);
            SyncFuture syncFuture = SyncFutureManager.get(slaveSyncRespDTO.getMsgId());
            // 从节点同步完成，对应的异步任务状态设置为Done
            if (syncFuture != null) {
                syncFuture.setResp(tcpMsg);
            }
        } else if (BrokerEventCode.CONSUME_LATER_MSG.getCode() == code) {
            ConsumeMsgRetryReqDTO consumeMsgRetryReqDTO = JSON.parseObject(body, ConsumeMsgRetryReqDTO.class);
            ConsumeMsgRetryEvent consumeMsgRetryEvent = new ConsumeMsgRetryEvent();
            consumeMsgRetryEvent.setMsgId(consumeMsgRetryReqDTO.getMsgId());
            consumeMsgRetryEvent.setConsumeMsgRetryReqDTO(consumeMsgRetryReqDTO);
            event = consumeMsgRetryEvent;
        }
        if (event != null) {
            event.setContext(channelHandlerContext);
            eventBus.publish(event);
        }
    }

    private void checkTopicIsCreate(String topic, ChannelHandlerContext context) throws InterruptedException {
        if (!CommonCache.getMqTopicModelMap().containsKey(topic)) {
            CreateTopicReqDTO createTopicReqDTO = new CreateTopicReqDTO();
            createTopicReqDTO.setTopic(topic);
            createTopicReqDTO.setMsgId(UUID.randomUUID().toString());
            createTopicReqDTO.setQueueSize(3);
            CreateTopicEvent createTopicEvent = new CreateTopicEvent();
            createTopicEvent.setCreateTopicReqDTO(createTopicReqDTO);
            createTopicEvent.setMsgId(createTopicReqDTO.getMsgId());
            createTopicEvent.setContext(context);
            Listener createTopicListener = eventBus.getListenerByClass(CreateTopicEvent.class);
            try {
                createTopicListener.onReceive(createTopicEvent);
                LOGGER.info("topic create success");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        LOGGER.error("error is :", cause);
    }


    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        //链接断开的时候，从重平衡池中移除
        Object reqId = ctx.attr(AttributeKey.valueOf("consumer-reqId")).get();
        if (reqId == null) {
            return;
        }
        CommonCache.getConsumerInstancePool().removeFromInstancePool(String.valueOf(reqId));
    }

    //@Override
    //public void channelActive(ChannelHandlerContext ctx) throws Exception {
    //    LOGGER.info("new connection build");
    //}
}
